package org.example.snow.demo5;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Properties;

/**
 * @author snowsong
 */
public class StartApp {
    private static final String REDIS_SERVER = "0.0.0.0";

    private static final Integer REDIS_PORT = 6379;

    public static void main(String[] args) throws Exception {
        // 初始化
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置 Kafka 客户端的连接参数
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "172.16.1.173:9092");
        FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer<>("xue",
                new SimpleStringSchema(), properties);
        DataStreamSource dataStreamSource = env.addSource(flinkKafkaConsumer);
        // 将接收的数据映射为二元组
        SingleOutputStreamOperator<Tuple2<String, String>> wordData = dataStreamSource.map(new MapFunction<String, Tuple2<String, String>>() {
            /**
             * 将输入的字符串映射为 Tuple2 对象。
             *
             * @param value 输入的字符串
             * @return 一个包含两个元素的 Tuple2 对象，第一个元素为 "l_words"，第二个元素为输入的字符串
             * @throws Exception 如果发生异常，则抛出该异常
             */
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                return new Tuple2<>("l_words", value);
            }
        });
        // 配置 Redis 连接池，设置 Redis 服务器地址和端口并构建对象
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig
                .Builder()
                .setHost(REDIS_SERVER)
                .setPort(REDIS_PORT)
                .build();
        // 创建 RedisSink 对象，用于将数据写入 Redis
        RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(conf, new MyRedisMapper());
        // 将 RedisSink 添加到数据流中，作为数据的接收端
        wordData.addSink(redisSink);
        env.execute();
    }

    /**
     * MyRedisMapper 类用于将 Flink 数据流中的 Tuple2 对象映射到 Redis 命令中。
     * 它实现了 RedisMapper 接口，用于自定义 Redis 数据的映射规则。
     */
    public static class MyRedisMapper implements RedisMapper<Tuple2<String,String>> {

        /**
         * 获取当前命令的描述信息。
         *
         * @return 返回Redis命令的描述信息对象，其中包含了命令的类型为LPUSH。
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        /**
         * 从给定的Tuple2数据中获取键。
         *
         * @param data 一个包含两个字符串元素的Tuple2对象
         * @return 返回Tuple2对象的第一个元素，即键
         */
        @Override
        public String getKeyFromData(Tuple2<String,String> data) {
            return data.f0;
        }

        /**
         * 从给定的元组中获取第二个元素的值。
         *
         * @param data 一个包含两个字符串元素的元组
         * @return 元组中的第二个元素的值
         */
        @Override
        public String getValueFromData(Tuple2<String,String> data) {
            return data.f1;
        }
    }

}